package org.study.canal.api;

import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 参考 https://blog.csdn.net/wu2374633583/article/details/120886666
 * 参考 https://tech.souyunku.com/?p=38015
 *
 * @author Administrator
 * @date 2021-05-20
 */
public class CanalTest {
    private static Logger log = LoggerFactory.getLogger(CanalTest.class);

    private static final int BATCH_SIZE = 1000;
    private static final String HOSTNAME = "127.0.0.1";
    private static final int PORT = 11111;
    private static final String DESTINATION = "example";
    private static final String USERNAME = "";
    private static final String PASSWORD = "";

    public static void main(String[] args) {
        String hostIp = AddressUtils.getHostIp();
        String hostName = AddressUtils.getHostName();
        InetAddress hostAddress = AddressUtils.getHostAddress();
        log.info("hostIp={}, hostName={}, hostAddress={}", hostIp, hostName, JSON.toJSONString(hostAddress));

        // 创建连接
        InetSocketAddress address = new InetSocketAddress(HOSTNAME, PORT);
        CanalConnector connector = CanalConnectors.newSingleConnector(address, DESTINATION, USERNAME, PASSWORD);
        long emptyCount = 0;
        try {
            // 打开连接
            connector.connect();
            // 指定要订阅的数据库表、表，格式为：数据库.表名,数据库.表名。常见例子：
            // 1、所有数据库下的所有表：.* 或 .*\\..*
            // 2、canal 数据库下所有表：canal\\..*
            // 3、canal 数据库下所有以 canal 开头的表：canal\\.canal.*
            // 4、canal 数据库下的 t_test 表：canal\\.t_test
            // 5、使用组合规则(逗号分隔)：canal\\..*,test.t_test1,demo.t_demo
            connector.subscribe(".*\\..*");
            // 回滚到未进行 ack 的地方，下次 fetch 时可以从最后一个没有 ack 的地方开始
            connector.rollback();
            long totalEmptyCount = Long.MAX_VALUE;
            while (emptyCount < totalEmptyCount) {
                // 获取指定数量的数据：尝试从 binlog 中拉取 batchSize 条记录，有多少取多少
                Message message = connector.getWithoutAck(BATCH_SIZE);
                // 获取批量ID
                long batchId = message.getId();
                // 获取批量的数量
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    // 如果没有数据
                    emptyCount++;
                    //log.debug("emptyCount：{}", emptyCount);
                    try {
                        // 线程休眠 1s
                        TimeUnit.SECONDS.sleep(1L);
                    } catch (InterruptedException e) {
                    }
                } else {
                    // 如果有数据，处理数据
                    emptyCount = 0;
                    //log.debug("message[batchId={}, size={}]", batchId, size);
                    printEntry(message.getEntries());
                }
                // 提交确认：对 batchId 进行确认，确认之后，小于等于此 batchId 的 Message 都会被确认
                connector.ack(batchId);
                // 处理失败，回滚数据，后续重新获取数据
                // connector.rollback(batchId);
            }
            log.warn("empty too many times, exit");
        } finally {
            connector.disconnect();
        }
    }

    /**
     * 打印 Canal Server 解析 binlog 获得的实体类信息
     *
     * @param entryList
     */
    private static void printEntry(List<CanalEntry.Entry> entryList) {
        for (CanalEntry.Entry entry : entryList) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                // 开启、关闭事务的实体类型：跳过
                continue;
            }
            CanalEntry.RowChange rowChange = null;
            try {
                // 获取发生变化的数据：RowChange 对象包含了一行数据变化的所有特征
                // 如 isDdl 表示是否为 ddl 操作，beforeColumns 表示变更前的数据字段，afterColumns 表示变更后的数据字段等
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("解析数据异常，当前数据：" + entry.toString(), e);
            }
            // 获取事件类型：INSERT、UPDATE、DELETE 类型
            CanalEntry.EventType eventType = rowChange.getEventType();
            log.debug("判断是否为 DDL 语句。isDdl={}，sql={}", rowChange.getIsDdl(), rowChange.getSql());

            CanalEntry.Header header = entry.getHeader();
            String formatMsg = String.format("logfileName=%s，logfileOffset=%s，schemaName=%s，tableName=%s，eventType=%s",
                    // 获取发生变化的 binlog 名称
                    header.getLogfileName(),
                    // 获取发生变化的 binlog 偏移量
                    header.getLogfileOffset(),
                    // 获取发生变化的数据库名称
                    header.getSchemaName(),
                    // 获取发生变化的表名称
                    header.getTableName(),
                    // 获取事件类型
                    eventType);
            log.info(formatMsg);

            // 获取 RowChange 对象里的每一行数据
            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                // 获取变更前的数据
                List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                // 获取变更后的数据
                List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                if (eventType == CanalEntry.EventType.INSERT) {
                    // 如果是 INSERT 语句
                    printColumn(afterColumnsList);
                } else if (eventType == CanalEntry.EventType.DELETE) {
                    // 如果是 DELETE 语句
                    printColumn(beforeColumnsList);
                } else if (eventType == CanalEntry.EventType.UPDATE) {
                    // 如果是 UPDATE 语句
                    System.out.println("【before update】");
                    printColumn(beforeColumnsList);
                    System.out.println("【after update】");
                    printColumn(afterColumnsList);
                } else {
                    // 其他语句
                    log.warn("其他(除 INSERT、DELETE、UPDATE 外)类型的事件");
                }
            }
        }
    }

    private static void printColumn(List<CanalEntry.Column> columnList) {
        for (CanalEntry.Column column : columnList) {
            System.out.println(column.getName() + "：" + column.getValue() + "，update=" + column.getUpdated());
        }
    }
}
